package com.fadis.cmd;

import com.fadis.data.FHyperLoglog;
import com.fadis.FadisException;
import com.fadis.data.hyperloglog.CardinalityMergeException;
import com.fadis.data.hyperloglog.HyperLogLog;
import com.fadis.db.FadisDB;
import com.fadis.protocol.LongNumber;
import com.fadis.protocol.SimpleString;

import java.util.ArrayList;
import java.util.List;

public class CmdHyperLogLog {

    // PFADD key element [element …]
    // 如果 HyperLogLog 的内部储存被修改了， 那么返回 1 ， 否则返回 0
    public static final class pfadd implements CmdHandler{

        @Override
        public String run(FadisActionFrame frame) throws FadisException {
            String key = frame.getArg(1);
            List<String> elements = new ArrayList<>();
            for (int i = 2; i < frame.getArgsCount(); i++) {
                elements.add(frame.getArg(i));
            }

           
            HyperLogLog data = FadisDB.tryToGetOrCreateHyperLogLog( key, true);

            boolean isChanged = false;
            for (int i = 0; i < elements.size(); i++) {
                if(data.offer(elements.get(i))) isChanged = true;
            }
            return LongNumber.encodeInt(isChanged ? 1 : 0);
        }
    }

    // PFCOUNT key [key …]   求基数
    // 多个键时，会临时合并所有key，算出一个基数返回
    // 返回基数
    public static final class pfcount implements CmdHandler{

        @Override
        public String run(FadisActionFrame frame) throws FadisException {
            List<String> keys = new ArrayList<>();
            for (int i = 1; i < frame.getArgsCount(); i++) {
                keys.add(frame.getArg(i));
            }

            if(keys.size() == 1){
                HyperLogLog data = FadisDB.tryToGetOrCreateHyperLogLog( keys.get(0), false);
                long count = data == null ? 0 : data.cardinality();
                return LongNumber.encodeLong(count);
            }else{
                HyperLogLog tempLog = FHyperLoglog.create();
                for (int i = 0; i < keys.size(); i++) {
                    HyperLogLog data = FadisDB.tryToGetOrCreateHyperLogLog( keys.get(i), false);
                    if(data != null) {
                        try {
                            tempLog.addAll(data);
                        } catch (CardinalityMergeException e) {
                            throw new FadisException(5002, "合并HyperLogLog时出错：" + e.getMessage());
                        }
                    }
                }
                long count = tempLog.cardinality();
                return LongNumber.encodeLong(count);
            }

        }
    }

    // PFMERGE destkey sourcekey [sourcekey …]   合并并存储到destkey，往已有的destkey上叠加，而非清空，并且就算合并结果是空，也会生成一个destkey
    // 返回 OK 。
    public static final class pfmerge implements CmdHandler{

        @Override
        public String run(FadisActionFrame frame) throws FadisException {
            String destkey = frame.getArg(1);
            List<String> srcs = new ArrayList<>();
            for (int i = 2; i < frame.getArgsCount(); i++) {
                srcs.add(frame.getArg(i));
            }

            HyperLogLog destLog = FadisDB.tryToGetOrCreateHyperLogLog( destkey, true);
            for (int i = 0; i < srcs.size(); i++) {
                HyperLogLog data = FadisDB.tryToGetOrCreateHyperLogLog( srcs.get(i), false);
                if(data != null) {
                    try {
                        destLog.addAll(data);
                    } catch (CardinalityMergeException e) {
                        throw new FadisException(5002, "合并HyperLogLog时出错：" + e.getMessage());
                    }
                }
            }
            return SimpleString.encodedSimpleString("OK");
        }
    }
}
